package raft

//
// this is an outline of the API that raft must expose to
// the service (or tester). see comments below for
// each of these functions for more details.
//
// rf = Make(...)
//   create a new Raft server.
// rf.Start(command interface{}) (count, Term, isleader)
//   start agreement on a new Log entry
// rf.GetState() (Term, isLeader)
//   ask a Raft for its current Term, and whether it thinks it is leader
// ApplyMsg
//   each Time a new entry is committed to the Log, each Raft peer
//   should send an ApplyMsg to the service (or tester)
//   in the same server.
//

import (
	"bytes"
	"fmt"
	"math/rand"
	"mit6.824/labgob"
	"mit6.824/labrpc"
	"runtime"
	//	"bytes"
	"sync"
	"sync/atomic"
	"time"
	//	"6.824/labgob"
)

//
// as each Raft peer becomes aware that successive Log Entries are
// committed, the peer should send an ApplyMsg to the service (or
// tester) on the same server, via the ApplyCh passed to Make(). set
// CommandValid to true to indicate that the ApplyMsg contains a newly
// committed Log entry.
//
// in part 2D you'll want to send other kinds of messages (e.g.,
// snapshots) on the ApplyCh, but set CommandValid to false for these
// other uses.
//
type ApplyMsg struct {
	CommandValid bool
	Command      interface{}
	CommandIndex int

	// For 2D:
	SnapshotValid bool
	Snapshot      []byte
	SnapshotTerm  int
	SnapshotIndex int
}

type LogEntry struct {
	Term    int         //任期号
	Index   int         //索引
	Command interface{} //命令
}

func (l *LogEntry) string() string {
	return fmt.Sprintf("[Term=%v,Command=%v]", l.Term, l.Command)
}

const (
	FOLLOWER  = 0
	CANDIDATE = 1
	LEADER    = 2

	HeartBeatTimeOut = 500                    //心跳超时
	HeartBeatRate    = 100                    //心跳频率
	ElectionTimeOut  = 300                    // 选举超时
	LoopInternal     = time.Millisecond * 100 //长时间循环休眠时间

	CommitLogName = "raft.Log" //持久化日志文件

	RequestVote   = 1
	AppendEntries = 2
)

type RaftTask interface {
	Run()
	WaitForDown()
	SetDown()
}

type RaftTaskRequestVote struct {
	taskType int
	raft     *Raft
	args     *RequestVoteArgs
	reply    *RequestVoteReply
	cond     sync.Cond
}

type RaftTaskAppendEntries struct {
	taskType int
	raft     *Raft
	args     *AppendEntriesArgs
	reply    *AppendEntriesReply
	cond     sync.Cond
}

// RaftTaskQueue 任务队列，存储收到的RPC任务
type RaftTaskQueue struct {
	task chan RaftTask
}

func (r *RaftTaskQueue) Pop() RaftTask {
	return <-r.task
}

func (r *RaftTaskQueue) Push(task RaftTask) {
	r.task <- task
}

func (r *RaftTaskQueue) Execute() {
	task := r.Pop()
	task.Run()
	task.SetDown()
}

func NewRaftTaskRequestVote(id int, raft *Raft, args *RequestVoteArgs, reply *RequestVoteReply) *RaftTaskRequestVote {
	task := RaftTaskRequestVote{
		taskType: id,
		raft:     raft,
		args:     args,
		reply:    reply,
	}
	task.cond.L = &sync.Mutex{}
	return &task
}
func NewRaftTaskAppendEntries(id int, raft *Raft, args *AppendEntriesArgs, reply *AppendEntriesReply) *RaftTaskAppendEntries {
	task := RaftTaskAppendEntries{
		taskType: id,
		raft:     raft,
		args:     args,
		reply:    reply,
	}
	task.cond.L = &sync.Mutex{}
	return &task
}

func (rt *RaftTaskRequestVote) Run() {
	rf := rt.raft
	args := rt.args
	reply := rt.reply
	reply.VoteGranted = false
	reply.Term = rf.currentTerm
	rf.lock()
	//直接拒绝任期小的
	if args.Term < rf.currentTerm {
		rf.unlock()
		//rf.Print(fmt.Sprintf("对方%v任期%v小，直接拒绝", Args.CandidateId, Args.Term))
		return
	}
	if args.Term > rf.currentTerm {
		reply.Term = args.Term
		if rf.LogManager.IsLogNewer(args.LastLogIndex, args.LastLogTerm) {
			reply.VoteGranted = true
			//rf.Print(fmt.Sprintf("对方%v任期%v一样，但是对方日志新，投给他", Args.CandidateId, Args.Term))
			rf.toFollower(args.Term, args.CandidateId)
		} else {
			reply.VoteGranted = false
			rf.toFollower(args.Term, -1)
		}
		//rf.Print(fmt.Sprintf("对方%v任期%v大，给他投票", Args.CandidateId, Args.Term))
	} else if rf.voteFor == -1 || rf.voteFor == args.CandidateId {
		if rf.LogManager.IsLogNewer(args.LastLogIndex, args.LastLogTerm) {
			reply.VoteGranted = true
			//rf.Print(fmt.Sprintf("对方%v任期%v一样，但是对方日志新，投给他", Args.CandidateId, Args.Term))
			rf.toFollower(args.Term, args.CandidateId)
		}
	}
	rf.unlock()
}

func (rt *RaftTaskRequestVote) WaitForDown() {
	rt.cond.L.Lock()
	rt.cond.Wait()
	rt.cond.L.Unlock()
}

func (rt *RaftTaskRequestVote) SetDown() {
	rt.cond.L.Lock()
	rt.cond.Broadcast()
	rt.cond.L.Unlock()
}

func (rt *RaftTaskAppendEntries) Run() {
	args := rt.args
	rf := rt.raft
	rf.timer.Reset()
	rf.lock()
	rt.reply.Term = rf.currentTerm
	/*直接拒绝小的term*/
	if args.Term < rf.currentTerm {
		//rf.Print(fmt.Sprintf("收到了领导人%v的心跳包,对方term小", Args.LeaderId))
		rt.reply.SetFalse()
		rf.unlock()
		return
	}

	/*如果对方term大，需要变成跟随者*/
	/*或者自己在选举但是收到了其他领导人的心跳，也需要变成跟随者*/
	if args.Term > rf.currentTerm || args.Term == rf.currentTerm && rf.state.IsCandidate() {
		//rf.Print(fmt.Sprintf("收到了领导人%v的心跳包,对方term大", Args.LeaderId))
		rt.reply.Term = args.Term
		rf.toFollower(args.Term, args.LeaderId)
	}

	/*拒绝不一致的日志*/
	if !rf.LogManager.CheckLogConsistency(args.PreLogIndex, args.PreLogTerm, rt.reply) {
		rt.reply.SetFalse()
		rf.Print(fmt.Sprintf("【%v】%v,日志不一致%v %v %v,当前日志%v", args.Index, rt.reply.Success, args.PreLogIndex, args.PreLogTerm, args.LeaderCommit, rf.LogManager.logPrint()))
		rf.unlock()
		return
	}
	/*以下是日志一致的情况*/
	rt.reply.SetTrue()
	if len(args.Entries) == 0 {
		rf.Print(fmt.Sprintf("【%v】收到了领导人%v的心跳包,对方term相等，正常接收", args.Index, args.LeaderId))
	} else {
		rf.Print(fmt.Sprintf("【%v】收到了来自%v的副本[term=%v,command=%v,leaderCommit=%v]", args.Index, args.LeaderId, args.Entries[0].Term, args.Entries[0].Command, args.LeaderCommit))
	}

	rf.LogManager.lock.Lock()
	rf.LogManager.updateLog(args.PreLogIndex, args.Entries)
	/* 2B:判断并更新commitIndex*/
	if args.LeaderCommit > rf.LogManager.CommitIndex {
		//old := rf.LogManager.CommitIndex
		//rf.Print(fmt.Sprintf("领导人的commitIndex=%v更大，preLogIndex=%v,准备commit", Args.LeaderCommit, Args.PreLogTerm))
		if args.PreLogIndex < args.LeaderCommit {
			rf.LogManager.CommitIndex = args.PreLogIndex
		} else {
			rf.LogManager.CommitIndex = args.LeaderCommit
		}
		//rf.Print(fmt.Sprintf("由%v更新为%v", old, rf.LogManager.CommitIndex))
		rf.LogManager.commitLog()
	}
	rf.LogManager.lock.Unlock()
	rf.unlock()
}

func (rt *RaftTaskAppendEntries) WaitForDown() {
	rt.cond.L.Lock()
	rt.cond.Wait()
	rt.cond.L.Unlock()
}

func (rt *RaftTaskAppendEntries) SetDown() {
	rt.cond.L.Lock()
	rt.cond.Broadcast()
	rt.cond.L.Unlock()
}

type ElectionManager struct {
	raft         *Raft
	args         *RequestVoteArgs
	replies      []RequestVoteReply
	winCount     int
	successCount chan bool
	isDone       bool
	isDoneChan   chan bool
	lock         sync.Mutex
}

func NewElectionManager(raft *Raft, args *RequestVoteArgs) *ElectionManager {
	return &ElectionManager{
		raft:         raft,
		args:         args,
		replies:      make([]RequestVoteReply, len(raft.peers)),
		winCount:     len(raft.peers) / 2,
		successCount: make(chan bool, len(raft.peers)),
	}
}

func (e *ElectionManager) start() {
	//e.raft.Print("开始选举！")
	rf := e.raft
	for i := 0; i < len(rf.peers); i++ {
		if !rf.state.IsCandidate() || rf.killed() {
			return
		}
		if i == rf.me {
			continue
		}
		go e.handleReply(i)
	}
}
func (e *ElectionManager) handleReply(server int) {
	rf := e.raft
	args := e.args
	reply := e.replies[server]
	ok := rf.sendRequestVote(server, args, &reply)
	if !ok {
		//rf.Print("没收到回应！")
	}

	if !ok || rf.killed() || !rf.state.IsCandidate() {
		return
	}
	if reply.VoteGranted {
		//str := "收到了" + strconv.Itoa(server) + "的投票"
		//rf.Print(str)
		e.successCount <- true
	}
	rf.lock()
	/* 这里需要斟酌一下 >=还是> */
	if reply.Term > rf.currentTerm {
		//str := strconv.Itoa(server) + "的任期更大，请求投票被拒绝"
		//rf.Print(str)
		rf.toFollower(reply.Term, server)
	}
	rf.unlock()
}

func (e *ElectionManager) wait() {
	//e.raft.Print("等待选举结果。。。")
	count := 200
	divDuration := e.raft.electionTimeOut / count
	for i := 0; i < count; i++ {
		if len(e.successCount)+1 > e.winCount {
			//e.raft.Print("选举成功！")
			e.raft.toLeader()
			return
		}
		if !e.raft.state.IsCandidate() {
			//e.raft.Print("选举提前结束！")
			return
		}
		time.Sleep(time.Millisecond * time.Duration(divDuration))
	}
	//e.raft.Print("选举超时！！")
	return
}

type ServerState struct {
	state int
}

func (s *ServerState) IsFollower() bool {
	return s.state == FOLLOWER
}
func (s *ServerState) IsCandidate() bool {
	return s.state == CANDIDATE
}
func (s *ServerState) IsLeader() bool {
	return s.state == LEADER
}

type RaftTimer struct {
	TimeOut bool
}

func (r *RaftTimer) Reset() {
	r.TimeOut = true
}

type RaftLogManager struct {
	Log []LogEntry //日志
	/*不稳定的状态*/
	CommitIndex int //最新的已提交日志索引
	LastApplied int //最新的应用到状态机的日志索引

	/*leader的不稳定状态*/
	NextIndex  []int //对于每个服务器，保存下一个要发送给它的日志索引，初始化为leader最后一个日志条目的索引加一
	MatchIndex []int //对于每个服务器，保存已经被复制的最新日志条目
	Success    chan int

	/* 2B */
	ApplyCh   chan ApplyMsg //提交通道
	CheckChan chan string

	raft           *Raft
	lock           sync.Mutex
	HeartBeatCount int //统计心跳次数，用于打印输出
}

func (rlm *RaftLogManager) len() int {
	return len(rlm.Log)
}

func (rlm *RaftLogManager) IndexFirst() LogEntry {
	if rlm.len() == 1 {
		return rlm.Log[0]
	}
	return rlm.Log[1]
}

func (rlm *RaftLogManager) IndexLast() LogEntry {
	if rlm.len() == 1 {
		return rlm.Log[0]
	}
	return rlm.Log[rlm.len()-1]
}

//根据索引获取Entry
func (rlm *RaftLogManager) Index(index int) LogEntry {
	return rlm.Log[index]
}

// Term 二分查找,返回一个term的起始索引和结束索引,如果没有则均为0
func (rlm *RaftLogManager) Term(term int) (int, int) {
	if term > rlm.IndexFirst().Term || term < rlm.IndexLast().Term {
		return 0, 0
	}
	if term == -1 {
		return 0, 0
	}
	left, right := 1, rlm.len()-1
	for {
		if left < right {
			break
		}
		mid := (left + right) / 2
		if term > rlm.Index(mid).Term {
			left = mid + 1
		} else if term < rlm.Index(mid).Term {
			right = mid - 1
		} else {
			left = mid
			right = mid
			for ; left > 0 && term != rlm.Index(left).Term; left-- {
			}
			for ; right < rlm.len() && term != rlm.Index(right).Term; right++ {
			}
			return left + 1, right - 1
		}
	}
	return 0, 0
}

func (rlm *RaftLogManager) logPrint() string {
	var str string
	str = "{"
	for _, entry := range rlm.Log {
		str += entry.string()
	}
	str += "}"
	return str
}

func NewRaftLogManager(rf *Raft, applyCh chan ApplyMsg) *RaftLogManager {
	log := make([]LogEntry, 1)
	log[0].Command = nil
	log[0].Term = -1
	RaftNum := len(rf.peers)
	//fmt.Printf("%v变成领导人 \n", rf.me)
	NextIndex := make([]int, RaftNum)
	return &RaftLogManager{
		Log:         log,
		raft:        rf,
		CommitIndex: 0,
		LastApplied: 0,
		NextIndex:   NextIndex,
		ApplyCh:     applyCh,
		CheckChan:   make(chan string),
	}
}

//心跳
func (r *RaftLogManager) heartbeat() {
	//rf.Print("开始发送心跳包")
	r.HeartBeatCount = 1
	for {
		r.lock.Lock()
		r.HeartBeatCount++
		rf := r.raft
		rf.Print(fmt.Sprintf("===========开始发送心跳包[%v]===========", r.HeartBeatCount))
		r.Success = make(chan int, len(r.raft.peers))
		r.lock.Unlock()
		for i, _ := range rf.peers {
			if !rf.state.IsLeader() || rf.killed() {
				return
			}
			if i == rf.me {
				continue
			}
			go r.handleReply(i)
		}
		r.waitReplicated()
		time.Sleep(time.Millisecond * time.Duration(HeartBeatRate)) //每秒发送10次心跳
	}
}

func (r *RaftLogManager) waitReplicated() {
	//r.raft.Print(fmt.Sprintf("[%v]等待复制完成,当前chan里面有%v个", count, len(r.Success)))
	for {
		if len(r.Success)+1 > len(r.raft.peers)/2 {
			//l := len(r.Success) + 1
			//id := <-r.Success
			//r.raft.Print(fmt.Sprintf("[%v]当前副本拥有%v份，超过一半,有一个是%v", count, l, id))
			break
		}
		if !r.raft.state.IsLeader() {
			r.raft.Print("身份转变，退出等待复制！")
			return
		}
	}
	r.lock.Lock()

	oldCommitIndex := r.CommitIndex
	//r.raft.Print("准备更新commitIndex！")
	r.raft.Print(fmt.Sprintf("[%v] match[%v,%v,%v]", r.HeartBeatCount, r.MatchIndex[0], r.MatchIndex[1], r.MatchIndex[2]))
	//如果存在一个满足N > commitIndex的 N，
	//并且大多数的matchIndex[i] ≥ N成立，
	//并且log[N].term == currentTerm成立，那么令 CommitIndex 等于这个 N
	for N := len(r.Log) - 1; N >= r.CommitIndex+1; N-- {
		count := 1
		for i, v := range r.MatchIndex {
			if i == r.raft.me {
				continue
			}
			if v >= N {
				count += 1
			}
		}
		if count > len(r.MatchIndex)/2 && r.Log[N].Term == r.raft.currentTerm {
			r.CommitIndex = N
			r.raft.Print(fmt.Sprintf("[%v]大多数副本都复制了到了索引%v，更新commmitIndex到%v", r.HeartBeatCount, N, N))
			break
		}
	}
	if r.CommitIndex > oldCommitIndex {
		r.commitLog()
	}
	r.lock.Unlock()
	//r.raft.Print(fmt.Sprintf("[%v]复制完成", count))
}
func (r *RaftLogManager) handleReply(i int) {

	r.lock.Lock()
	Args := r.NewAppendEntriesArgs(i)
	Reply := AppendEntriesReply{}
	var command string
	if len(Args.Entries) > 0 {
		for _, entry := range Args.Entries {
			command += entry.string()
		}
		r.raft.Print(fmt.Sprintf("【%v】准备向%v发送%v份副本[%v]", Args.Index, i, len(Args.Entries), command))
	} else {
		r.raft.Print(fmt.Sprintf("【%v】准备向%v发送心跳", Args.Index, i))
	}

	r.lock.Unlock()
	var ok bool = false
	for !ok || !Reply.IsSuccess() {
		ok = r.raft.peers[i].Call("Raft.AppendEntries", &Args, &Reply)
		r.raft.Print(fmt.Sprintf("【%v】服务器%v的返回结果是success=%v，term=%v", Args, i, Reply.Success, Reply.Term))
		if !r.raft.state.IsLeader() {
			r.raft.Print("身份转变，退出处理日志RPC回复！")
			return
		}
		if ok {
			//对方term更大，直接变为跟随者
			if Reply.Term > r.raft.currentTerm {
				r.raft.Print(fmt.Sprintf("向服务器%v发送日志失败，对方term大，变成跟随者！", i))
				r.raft.toFollower(Reply.Term, i)
				return
			}
			if !Reply.IsSuccess() {
				r.raft.Print(fmt.Sprintf("【%v】向服务器%v发送日志失败，准备重试", Args.Index, i))
				r.lock.Lock()
				_, lt := r.Term(Reply.XTerm)
				if lt != 0 {
					r.NextIndex[i] = r.Index(lt).Index + 1
				} else {
					r.NextIndex[i] = Reply.XIndex
				}
				Args = r.NewAppendEntriesArgs(i)
				Reply = AppendEntriesReply{}
				r.raft.Print(fmt.Sprintf("【%v】NextIndex[%v]减一变成了%v", Args.Index, i, r.NextIndex[i]))
				r.lock.Unlock()
			}
		} else {
			r.raft.Print(fmt.Sprintf("向服务器%v发送日志没有收到回应，准备重试", i))
		}
	}
	//r.raft.Print(fmt.Sprintf("向服务器%v发送日志[%v]成功", i, command))
	r.lock.Lock()
	r.NextIndex[i] += len(Args.Entries)
	r.MatchIndex[i] = r.NextIndex[i] - 1
	r.lock.Unlock()
	//r.raft.Print(fmt.Sprintf("[%v]NextIndex[%v]=%v MatchIndex[%v]=%v", count, i, r.NextIndex[i], i, r.MatchIndex[i]))
	r.Success <- i
}

func (r *RaftLogManager) appendLog(entry LogEntry) {
	//r.raft.Print(fmt.Sprintf("收到了客户端的命令并准备更新日志"))
	r.Log = append(r.Log, entry)
	r.raft.persist()
	//r.raft.Print(fmt.Sprintf("收到了客户端的命令并成功更新日志"))
}
func (r *RaftLogManager) updateLog(lastIndex int, entry []LogEntry) {
	r.Log = r.Log[:lastIndex+1]
	r.Log = append(r.Log, entry...)
	r.raft.persist()
	r.raft.Print(fmt.Sprintf("收到副本,成功更新日志为：%v！", r.logPrint()))
}
func (r *RaftLogManager) commitLog() {
	//r.raft.Print(fmt.Sprintf("当前日志是" + r.logPrint() + "\n"))

	for index := r.LastApplied + 1; index <= r.CommitIndex; index++ {
		entry := r.Log[index]
		applyMsg := ApplyMsg{true, entry.Command, index, false, []byte{}, 0, 0}
		r.ApplyCh <- applyMsg
		//r.raft.Print("成功向状态机提交命令" + entry.string())
	}
	r.LastApplied = r.CommitIndex
}

//判断传入的最后一条日志索引和任期号是否更新
//如果传入的更新，返回true，否则false
func (rf *RaftLogManager) IsLogNewer(lastIndex int, lastTerm int) bool {
	index, term := rf.GetLastLogState()
	if lastTerm > term || (lastTerm == term && lastIndex >= index) { //如果对方的日期更新
		return true
	} else {
		return false
	}
}

// CheckLogConsistency follower检查日志一致性
func (rf *RaftLogManager) CheckLogConsistency(preLogIndex int, preLogTerm int, reply *AppendEntriesReply) bool {
	rf.raft.Print(fmt.Sprintf("preLogIndex=%v,preLogTerm=%v", preLogIndex, preLogTerm))
	if preLogIndex <= 0 { //如果是心跳包，直接返回true
		return true
	}
	if preLogIndex < rf.len() && rf.Log[preLogIndex].Term == preLogTerm { //如果当前服务器与preLog处的日志条目一致，则为true
		//XTerm  int //在日志一致性检查时，当前follower冲突的日志条目的term，leader首先检查自己的日志中是否有这个XTerm,发送自己XTerm最后一条日志条目之后的日志
		//XIndex int //该Xterm下所属的第一条日志条目索引，如果leader发现自己没有这个XTerm就会设置nextIndex为这个XIndex
		//XLen   int //follower返回他的日志长度，如果leader检查发现与follower日志长度不一致，就检查
		return true
	} else {
		lastTerm := rf.Index(rf.len() - 1).Term
		reply.XTerm = lastTerm
		reply.XIndex, _ = rf.Term(lastTerm)
		return false
	}
}

//获取最后一条日志条目的任期号和索引
func (rf *RaftLogManager) GetLastLogState() (int, int) {
	return len(rf.Log) - 1, rf.Log[len(rf.Log)-1].Term
}

func (r *RaftLogManager) NewAppendEntriesArgs(i int) AppendEntriesArgs {
	preLogIndex, preLogTerm := r.GetPreLogState(i)
	var entries []LogEntry
	if r.NextIndex[i] >= len(r.Log) {
		entries = make([]LogEntry, 0)
	} else {
		entries = r.Log[r.NextIndex[i]:]
	}
	return AppendEntriesArgs{
		Term:         r.raft.currentTerm,
		PreLogIndex:  preLogIndex,
		PreLogTerm:   preLogTerm,
		Entries:      entries,
		LeaderId:     r.raft.me,
		LeaderCommit: r.CommitIndex,
		Index:        r.HeartBeatCount,
	}
}

func (r *RaftLogManager) GetPreLogState(i int) (int, int) {
	preLogIndex := -1
	preLogTerm := -1
	if r.NextIndex[i] > 0 {
		if r.NextIndex[i] > len(r.Log) {
			r.NextIndex[i] = len(r.Log)
		}
		preLogIndex = r.NextIndex[i] - 1
		preLogTerm = r.Log[preLogIndex].Term
	}
	return preLogIndex, preLogTerm
}

//
// A Go object implementing a single Raft peer.
//
type Raft struct {
	mu        sync.Mutex          // lock to protect shared access to this peer's state
	peers     []*labrpc.ClientEnd // RPC end points of all peers
	persister *Persister          // Object to hold this peer's persisted state
	me        int                 // this peer's count into peers[]
	dead      int32               // set by Kill()

	/* 永久状态 */
	currentTerm int //任期号
	voteFor     int //当前服务器投票的候选者Id，

	/*2A*/
	state            ServerState //服务器状态：0跟随者，1候选人，2领导人
	electionTimeOut  int         //选举超时时间
	heartBeatTimeOut int
	timer            RaftTimer //是否超时
	queue            *RaftTaskQueue

	/*2B*/
	LogManager RaftLogManager
}

func printCallerName() string {
	pc, _, _, _ := runtime.Caller(2)
	return runtime.FuncForPC(pc).Name()
}

func (r *Raft) Print(string2 string) {
	var state string
	if r.state.IsFollower() {
		state = "f"
	} else if r.state.IsCandidate() {
		state = "c"
	} else {
		state = "l"
	}
	fmt.Sprintf("[id=%v stete=%v term=%v vote=%v CommitIndex=%v LastApplied=%v logLength=%v] %v \n", r.me, state, r.currentTerm, r.voteFor, r.LogManager.CommitIndex, r.LogManager.LastApplied, len(r.LogManager.Log), string2)
}

func (rf *Raft) handleTask() {
	for {
		rf.queue.Execute()
		rf.queue.Execute()
	}
}

func (rf *Raft) lock() {
	//fmt.Printf("服务器%v在%v处加锁\n", rf.me, printCallerName())
	rf.mu.Lock()
}

func (rf *Raft) unlock() {
	//fmt.Printf("服务器%v在%v处解锁\n", rf.me, printCallerName())
	rf.mu.Unlock()
}

// return currentTerm and whether this server
// believes it is the leader.
func (rf *Raft) GetState() (int, bool) {
	return rf.currentTerm, rf.state.IsLeader()
}

func (rf *Raft) toFollower(term int, candidateId int) {
	//rf.Print("变成跟随者")
	rf.state = ServerState{state: FOLLOWER}
	rf.currentTerm = term
	rf.voteFor = candidateId
}
func (rf *Raft) toCandidate() {
	rf.state = ServerState{state: CANDIDATE}
	rf.voteFor = -1
	rf.Print("变成候选人")
}

//每次选举后转为领导人，并初始化相关参数
func (rf *Raft) toLeader() {
	//rf.Print("变成领导者")

	rf.state = ServerState{state: LEADER}
	rf.voteFor = rf.me
	RaftNum := len(rf.peers)
	//fmt.Printf("%v变成领导人 \n", rf.me)
	//NextIndex[]初始化为领导人最后一条日志的索引加一
	lastIndex, _ := rf.LogManager.GetLastLogState()
	rf.LogManager.NextIndex = make([]int, RaftNum)
	//rf.Print(fmt.Sprintf("初始化nextIndex为%v", lastIndex+1))
	for i := 0; i < RaftNum; i++ {
		rf.LogManager.NextIndex[i] = lastIndex + 1
	}
	//MatchIndex[]初始化为0
	rf.LogManager.MatchIndex = make([]int, RaftNum)

	//初始化replicated参数
	rf.LogManager.Success = make(chan int, RaftNum)
}

func (rf *Raft) doFollower() {
	rf.unlock()
	if !rf.checkHeartBeatTimeOut() {
		rf.lock()
		rf.toCandidate()
	} else {
		rf.lock()
	}
}

// 开启选举
func (rf *Raft) doCandidate() {
	r := rand.Intn(rf.electionTimeOut)
	//rf.Print(fmt.Sprintf("超时时间是%v", r))
	rf.unlock()
	time.Sleep(time.Millisecond * time.Duration(r))
	if !rf.state.IsCandidate() {
		rf.lock()
		return
	}
	rf.lock()
	rf.voteFor = rf.me
	rf.currentTerm += 1
	lastIndex, lastTerm := rf.LogManager.GetLastLogState()
	args := RequestVoteArgs{rf.currentTerm, rf.me, lastIndex, lastTerm}
	manager := NewElectionManager(rf, &args)
	rf.unlock()
	manager.start()
	manager.wait()
	rf.lock()
}

func (rf *Raft) doLeader() {
	rf.unlock()
	rf.LogManager.heartbeat()
	rf.lock()
}

//
// save Raft's persistent state to stable storage,
// where it can later be retrieved after a crash and restart.
// see paper's Figure 2 for a description of what should be persistent.
//
func (rf *Raft) persist() {
	// Your code here (2C).
	// Example:
	w := new(bytes.Buffer)
	e := labgob.NewEncoder(w)
	e.Encode(rf.currentTerm)
	e.Encode(rf.voteFor)
	e.Encode(rf.LogManager.Log)
	data := w.Bytes()
	rf.persister.SaveRaftState(data)
	rf.Print(fmt.Sprintf("持久化的数据是：[voteFor=%v,term=%v,log=%v", rf.voteFor, rf.currentTerm, rf.LogManager.logPrint()))

	////rf.Print("准备持久化：" + rf.LogManager.logPrint())
	//file, _ := os.Create(strconv.Itoa(rf.me) + ".json")
	////fmt.Printf("持久化日志")
	//for _, entry := range rf.LogManager.Log {
	//	enc := json.NewEncoder(file)
	//	enc.Encode(&entry)
	//}
	//rf.Print("持久化完成！")

}

//
// restore previously persisted state.
//
func (rf *Raft) readPersist(data []byte) {
	if data == nil || len(data) < 1 { // bootstrap without any state?
		return
	}
	// Your code here (2C).
	// Example:
	// r := bytes.NewBuffer(data)
	// d := labgob.NewDecoder(r)
	// var xxx
	// var yyy
	// if d.Decode(&xxx) != nil ||
	//    d.Decode(&yyy) != nil {
	//   error...
	// } else {
	//   rf.xxx = xxx
	//   rf.yyy = yyy
	// }

	r := bytes.NewBuffer(data)
	d := labgob.NewDecoder(r)
	var currentTerm int
	var voteFor int
	var Log []LogEntry
	if d.Decode(&currentTerm) != nil ||
		d.Decode(&voteFor) != nil ||
		d.Decode(&Log) != nil {
		rf.Print("读取异常！")
	} else {
		rf.voteFor = voteFor
		rf.currentTerm = currentTerm
		rf.LogManager.Log = Log
		rf.Print(fmt.Sprintf("读取到的数据是：[voteFor=%v,term=%v,log=%v", voteFor, currentTerm, rf.LogManager.logPrint()))
	}
}

//
// A service wants to switch to snapshot.  Only do so if Raft hasn't
// have more recent info since it communicate the snapshot on ApplyCh.
//
func (rf *Raft) CondInstallSnapshot(lastIncludedTerm int, lastIncludedIndex int, snapshot []byte) bool {

	// Your code here (2D).

	return true
}

// the service says it has created a snapshot that has
// all info up to and including count. this means the
// service no longer needs the Log through (and including)
// that count. Raft should now trim its Log as much as possible.
func (rf *Raft) Snapshot(index int, snapshot []byte) {
	// Your code here (2D).

}

//
// example RequestVote RPC arguments structure.
// field names must start with capital letters!
//
type RequestVoteArgs struct {
	// Your data here (2A, 2B).
	Term         int //任期
	CandidateId  int //候选人ID
	LastLogIndex int //候选人最后一条日志索引
	LastLogTerm  int //候选人最后一条日志任期
}

//
// example RequestVote RPC reply structure.
// field names must start with capital letters!
//
type RequestVoteReply struct {
	// Your data here (2A).
	Term        int  //投票人任期
	VoteGranted bool //投票人是否投票
}

//
// example RequestVote RPC handler.
// 处理请求投票的RPC
func (rf *Raft) RequestVote(args *RequestVoteArgs, reply *RequestVoteReply) {
	//rf.Print("收到了请求投票！")
	task := NewRaftTaskRequestVote(RequestVote, rf, args, reply)
	rf.queue.Push(task)
	task.WaitForDown()
	//str := fmt.Sprintf("对方%v的term=%v,RPC的处理结果是term=%v,granted=%v", Args.CandidateId, Args.Term, reply.Term, reply.VoteGranted)
	//rf.Print(str)
}

//
// example code to send a RequestVote RPC to a server.
// server is the count of the target server in rf.peers[].
// expects RPC arguments in Args.
// fills in *reply with RPC reply, so caller should
// pass &reply.
// the types of the Args and reply passed to Call() must be
// the same as the types of the arguments declared in the
// handler function (including whether they are pointers).
//
// The labrpc package simulates a lossy network, in which servers
// may be unreachable, and in which requests and Replies may be lost.
// Call() sends a request and waits for a reply. If a reply arrives
// within a timeout interval, Call() returns true; otherwise
// Call() returns false. Thus Call() may not return for a while.
// A false return can be caused by a dead server, a live server that
// can't be reached, a lost request, or a lost reply.
//
// Call() is guaranteed to return (perhaps after a delay) *except* if the
// handler function on the server side does not return.  Thus there
// is no need to implement your own timeouts around Call().
//
// look at the comments in ../labrpc/labrpc.go for more details.
//
// if you're having trouble getting RPC to work, check that you've
// capitalized all field names in structs passed over RPC, and
// that the caller passes the address of the reply struct with &, not
// the struct itself.
//
func (rf *Raft) sendRequestVote(server int, args *RequestVoteArgs, reply *RequestVoteReply) bool {
	ok := rf.peers[server].Call("Raft.RequestVote", args, reply)
	return ok
}

//
// the service using Raft (e.g. a k/v server) wants to start
// agreement on the next command to be appended to Raft's Log. if this
// server isn't the leader, returns false. otherwise start the
// agreement and return immediately. there is no guarantee that this
// command will ever be committed to the Raft Log, since the leader
// may fail or lose an election. even if the Raft instance has been killed,
// this function should return gracefully.
//
// the first return value is the count that the command will appear at
// if it's ever committed. the second return value is the current
// Term. the third return value is true if this server believes it is
// the leader.
//
func (rf *Raft) Start(command interface{}) (int, int, bool) {
	//rf.Print(fmt.Sprintf("收到了客户端的命令[%v]", command))
	index := -1
	term, isLeader := rf.GetState()

	// Your code here (2B).
	rf.lock()
	if isLeader {
		index = len(rf.LogManager.Log)
		rf.LogManager.lock.Lock()
		entry := LogEntry{rf.currentTerm, rf.LogManager.len(), command}
		rf.LogManager.appendLog(entry)
		rf.Print(fmt.Sprintf("初始化nextIndex为%v", index))
		for i := 0; i < len(rf.peers); i++ {
			rf.LogManager.NextIndex[i] = index
		}
		rf.Print(fmt.Sprintf("收到了客户端的命令[%v],当前日志是%v", command, rf.LogManager.logPrint()))
		rf.LogManager.lock.Unlock()
	}
	rf.unlock()
	//rf.Print(fmt.Sprintf("start返回"))
	return index, term, isLeader
}

//
// the tester doesn't halt goroutines created by Raft after each test,
// but it does call the Kill() method. your code can use killed() to
// check whether Kill() has been called. the use of atomic avoids the
// need for a lock.
//
// the issue is that long-running goroutines use memory and may chew
// up CPU Time, perhaps causing later tests to fail and generating
// confusing debug output. any goroutine with a long-running loop
// should call killed() to check whether it should stop.
//
func (rf *Raft) Kill() {
	atomic.StoreInt32(&rf.dead, 1)
	// Your code here, if desired.
}

func (rf *Raft) killed() bool {
	z := atomic.LoadInt32(&rf.dead)
	return z == 1
}

func (rf *Raft) checkState() {
	for {
		rf.lock()
		if rf.state.IsFollower() {
			rf.doFollower()
		} else if rf.state.IsCandidate() {
			rf.doCandidate()
		} else {
			rf.doLeader()
		}
		rf.unlock()
		//time.Sleep(LoopInternal)
	}
}

//心跳检测，收到心跳返回true，超时返回false
func (rf *Raft) checkHeartBeatTimeOut() bool {
	checkCount := 200
	divDuration := rf.heartBeatTimeOut / checkCount
	//每次检测休息一次，共200次
	for i := 0; i < checkCount; i++ {
		if rf.timer.TimeOut {
			rf.timer.TimeOut = false
			return true
		}
		time.Sleep(time.Millisecond * time.Duration(divDuration))
	}
	//rf.Print("心跳超时！")
	rf.timer.TimeOut = false
	return false
}

/*func (rf *Raft) checkElectionTimeOut() {
	for {
		Time.Sleep(LoopInternal)
		rf.lock()
		state := rf.state
		heartBeatTimeOut := rf.heartBeatTimeOut
		electionTimeOut := rf.electionTimeOut
		rf.unlock()
		if state != CANDIDATE {
			return
		}
		if Time.Since(heartBeatTimeOut) > HeartBeatTimeOut+electionTimeOut {
			//fmt.Printf("服务器%v选举超时\n", rf.me)
			go rf.doElection()
		}
	}
}*/

/*func (rf *Raft) selectTimeOut() {
	for {
		select {
		case <-rf.heartBeatReceiveChan: //心跳超时
			go func() {
				//fmt.Printf("服务器%v heartBeatReceiveChan start\n", rf.me)
				rf.lock()
				if rf.state == FOLLOWER {
					//a := <-rf.heartBeatTimeOutChan
					rf.heartBeatTimeOutChan = make(chan bool, 1)
					go func() { //进行超时计时，如果有下一个心跳到达则会取消，开启新的计时
						select {
						case <-Time.After(HeartBeatTimeOut):
							fmt.Printf("服务器%v心跳超时\n", rf.me)
							if rf.GetServerState() == FOLLOWER {
								go rf.doElection()
							}
						case <-rf.heartBeatTimeOutChan:
							//fmt.Printf("服务器%v检测到数据heartBeatTimeOutChan=%v插入,\n", rf.me, ok)
							rf.heartBeatTimeOutChan = nil
							//fmt.Printf("服务器%v插入数据heartBeatTimeOutChan=false,\n", rf.me)
							return
						}
					}()
				}
				//fmt.Printf("服务器%v heartBeatReceiveChan fin\n", rf.me)
				rf.unlock()
			}()
		case <-rf.electionStartChan: //收到开始选举通知
			go func() {
				rf.lock()
				defer rf.unlock()
				if rf.state == CANDIDATE {
					electionTimeOut := rf.electionTimeOut
					rf.electionTimeOutChan = make(chan bool, 1)
					go func() {
						select {
						case <-Time.After(electionTimeOut):
							fmt.Printf("[%v]服务器%v选举超时\n", Time.Now().Second(), rf.me)
							if rf.GetServerState() == CANDIDATE {
								go rf.doElection()
							}
						case <-rf.electionTimeOutChan:
							fmt.Printf("服务器%v取消选举超时", rf.me)
							rf.electionTimeOutChan = nil
							return
						}
					}()
				}
			}()
		default:
		}
	}
}
*/

type AppendEntriesArgs struct {
	Term        int
	LeaderId    int
	PreLogIndex int

	PreLogTerm int
	Entries    []LogEntry

	LeaderCommit int
	Index        int
}

// AppendEntriesReply 附加日志RPC
type AppendEntriesReply struct {
	Term    int
	Success bool
	//ConflictIndex int //冲突的任期中最小的索引
	XTerm  int //在日志一致性检查时，当前follower冲突的日志条目的term，leader首先检查自己的日志中是否有这个XTerm,发送自己XTerm最后一条日志条目之后的日志
	XIndex int //该Xterm下所属的第一条日志条目索引，如果leader发现自己没有这个XTerm就会设置nextIndex为这个XIndex
	XLen   int //follower返回他的日志长度，如果leader检查发现与follower日志长度不一致，就检查
	lock   sync.Mutex
}

func (r *AppendEntriesReply) IsSuccess() bool {
	//r.lock.Lock()
	//defer r.lock.Unlock()
	return r.Success
	//fmt.Printf("AppendEntriesReply set true in" + str + "\n")
}

func (r *AppendEntriesReply) SetTrue() {
	//r.lock.Lock()
	r.Success = true
	//r.lock.Unlock()
	//fmt.Printf("AppendEntriesReply set true in" + str + "\n")
}

func (r *AppendEntriesReply) SetFalse() {
	//r.lock.Lock()
	r.Success = false
	//r.lock.Unlock()
	//fmt.Printf("AppendEntriesReply set true in" + str + "\n")
}

//处理附加日志RPC
func (rf *Raft) AppendEntries(args *AppendEntriesArgs, reply *AppendEntriesReply) {
	rf.Print(fmt.Sprintf("【%v】收到%v的AppendEntries，当前reply=[%v,%v,commit=%v]", args.Index, args.LeaderId, reply.Term, reply.IsSuccess(), args.LeaderCommit))
	task := NewRaftTaskAppendEntries(AppendEntries, rf, args, reply)
	rf.queue.Push(task)
	task.WaitForDown()
	rf.Print(fmt.Sprintf("【%v】处理完%v的AppendEntries，当前reply=[%v,%v,commit=%v]", args.Index, args.LeaderId, reply.Term, reply.IsSuccess(), args.LeaderCommit))
}

type InstallSnapshotArgs struct {
	Term              int
	LeaderId          int
	LastIncludedIndex int
	LastIncludedTerm  int
	offset            int
	data              []byte
	done              bool
}

type InstallSnapshotReply struct {
	Term int
}

func (r *Raft) sendInstallSnapshot(server int, args *InstallSnapshotArgs, reply *InstallSnapshotReply) bool {
	return r.peers[server].Call("Raft.InstallSnapshot", args, reply)
}

func (r *Raft) InstallSnapshot(server int, args *InstallSnapshotArgs, reply *InstallSnapshotReply) {
	if r.currentTerm > args.Term {
		return
	}
}

//
// the service or tester wants to create a Raft server. the ports
// of all the Raft servers (including this one) are in peers[]. this
// server's port is peers[me]. all the servers' peers[] arrays
// have the same order. persister is a place for this server to
// save its persistent state, and also initially holds the most
// recent saved state, if any. ApplyCh is a channel on which the
// tester or service expects Raft to send ApplyMsg messages.
// Make() must return quickly, so it should start goroutines
// for any long-running work.
//
func Make(peers []*labrpc.ClientEnd, me int,
	persister *Persister, applyCh chan ApplyMsg) *Raft {
	rf := &Raft{}
	rf.peers = peers
	rf.persister = persister
	rf.me = me
	log := make([]LogEntry, 1)
	log[0].Term = -1
	log[0].Command = nil
	log[0].Index = -1
	/* 2B */
	rf.LogManager = *NewRaftLogManager(rf, applyCh)

	// Your initialization code here (2A, 2B, 2C).
	/* 2A */
	rf.state = ServerState{state: FOLLOWER} //初始化为跟随者 2A
	rf.voteFor = -1                         // 初始未投票 2A
	rf.currentTerm = 0                      // 初始任期为0 2A
	rf.electionTimeOut = ElectionTimeOut
	rf.heartBeatTimeOut = HeartBeatTimeOut
	rf.queue = &RaftTaskQueue{task: make(chan RaftTask)}

	// initialize from state persisted before a crash
	rf.readPersist(persister.ReadRaftState())

	// start ticker goroutine to start elections
	go rf.checkState()
	go rf.handleTask()
	//go rf.LogManager.checkLog()
	return rf
}

func runFuncName() string {
	pc := make([]uintptr, 1)
	runtime.Callers(2, pc)
	f := runtime.FuncForPC(pc[0])
	return f.Name()
}
